package com.hrt.flink.icebergoperate.unauto;

import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.hadoop.conf.Configuration;
import org.apache.iceberg.Table;
import org.apache.iceberg.catalog.Catalog;
import org.apache.iceberg.catalog.TableIdentifier;
import org.apache.iceberg.flink.actions.Actions;
import org.apache.iceberg.hadoop.HadoopCatalog;

/**
 *  合并data files
 */
public class FlinkIceberg3 {
    public static void main(String[] args) {
        ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();

        Configuration hadoopConf = new Configuration();
        Catalog catalog = new HadoopCatalog(hadoopConf,"hdfs://mycluster/flink_iceberg/");
        TableIdentifier name = TableIdentifier.of("icebergdb", "flink_iceberg_tbl");
        Table table = catalog.loadTable(name);

        //合并data files
        Actions.forTable(table)
                .rewriteDataFiles()
                .targetSizeInBytes(536870912L)
                .execute();
    }
}
